/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

/*
 * FirstReducer.java
 *
 * Created on Jul 9, 2010, 4:26:23 PM
 */
package homomorphism.list;

import homomorphism.HomomorphismBase;
import java.util.logging.Level;
import java.util.logging.Logger;
import utility.LongPair;
import java.io.IOException;
import java.lang.reflect.*;
import java.util.ArrayList;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

/**
 *
 * @author takeshi
 */
// mem org.apache.hadoop.mapred.lib.aggregate
//abstract
public class FirstReducer<InVal> extends MapReduceBase implements Reducer<LongPair, InVal, LongPair, InVal> {

    private static final Log LOG = LogFactory.getLog(FirstReducer.class);
    private static final String userClassPro = HomomorphismBase.UserClassNameProperty ;
    private static final String InValClassPro = HomomorphismBase.FoldingInValClassProperty ;//"mapred.input.value.class";
  //  private static final String OtValClassPro = HomomorphismBase.IndexKeyClassProperty;//"mapred.output.value.class";
    public static final int Mega = HomomorphismBase.Mega;
    private Class FilterType;
    private Class AggType;
    private String UserClassName;

    @Override
    public void configure(JobConf job) {

        super.configure(job);
        UserClassName = job.get(userClassPro);
        try {
            FilterType = Class.forName(job.get(InValClassPro));
        } catch (ClassNotFoundException ex) {
            Logger.getLogger(FirstReducer.class.getName()).log(Level.SEVERE, null, ex);
        }
        AggType = job.getMapOutputValueClass();

    }

    public InVal plus(InVal v1, InVal v2) throws Exception {

        Class cls = Class.forName(UserClassName);
        Class partypes[] = new Class[1];
        partypes[0] = v1.getClass();
        Method _plus = cls.getMethod("plus", partypes);
        Object arglist[] = new Object[2];
        arglist[0] = v1;
        arglist[1] = v2;
        Constructor ct = cls.getConstructor(); //partypes
        Object methobj = ct.newInstance();
        InVal rst = (InVal) _plus.invoke(methobj, arglist);
        return rst;
    }

    public InVal aggregator(ArrayList<InVal> values) throws Exception {

        Class cls = Class.forName(UserClassName);
        Class partypes[] = new Class[1];
       // ArrayList<? extends Writable> eee = new ArrayList();
        partypes[0] = values.getClass();
        Method _aggregator = null;
        InVal rst = null;
        try {
            _aggregator = cls.getMethod("aggregator", partypes);
        } catch (NoSuchMethodException ex) {
            LOG.debug(ex);
            int count = 0;
            for (InVal e : values) {
                if (count == 0) {
                    rst = e;
                } else {
                    rst = plus(rst, e);
                }
                count++;
            }
            return rst;
        }
        Object arglist[] = new Object[1];
        arglist[0] = values;
        Constructor ct = cls.getConstructor(); //partypes
        Object methobj = ct.newInstance();
         rst = (InVal) _aggregator.invoke(methobj, arglist);
        return rst;
    }


    public InVal aggregator(Iterator<InVal> values) throws Exception {

        Class cls = Class.forName(UserClassName);
        Class partypes[] = new Class[1];
        partypes[0] = Iterator.class;
        Method _aggregator = null;
        InVal rst = null;
        try {
            _aggregator = cls.getMethod("aggregator", partypes);
        } catch (NoSuchMethodException ex) {
            LOG.debug(ex);
            int count = 0;
            while ( values.hasNext()) {
                if (count == 0) {
                    rst = values.next();
                } else {
                    rst = plus(rst, values.next());
                }
                count++;
            }
            return rst;
        }
        Object arglist[] = new Object[1];
        arglist[0] = values;
        Constructor ct = cls.getConstructor(); //partypes
        Object methobj = ct.newInstance();
        rst = (InVal) _aggregator.invoke(methobj, arglist);
        return rst;
    }

    @Override
    public void reduce(LongPair key, Iterator<InVal> values, OutputCollector<LongPair, InVal> output, Reporter reporter)
            throws IOException {
        InVal rst = null;
        try {
            rst = aggregator(values);
        } catch (Exception ex) {
            Logger.getLogger(FirstReducer.class.getName()).log(Level.SEVERE, null, ex);
        }
        if(rst == null){
           NullPointerException enull =  new NullPointerException("aggregator result can not be null!");
            Logger.getLogger(FirstReducer.class.getName()).log(Level.SEVERE, null, enull);
            throw enull;
        }
        output.collect(key, rst);

    }
}
